package com.shujia.hbase.api;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;

import java.io.IOException;

public class Demo11HbaseToIndex {

    //读取hbase里面的数据

    public static class HbaseToIndexMap extends TableMapper<Text, NullWritable> {

        @Override
        protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {

            String rowkey = Bytes.toString(key.get());

            byte[] clazzByte = value.getValue("info".getBytes(), "clazz".getBytes());

            String clazz = Bytes.toString(clazzByte);


            //以班级和学号作为rowkey
            String line = clazz + "-" + rowkey;

            context.write(new Text(line), NullWritable.get());

        }
    }


    public static class HbaseToIndexReduce extends TableReducer<Text, NullWritable, NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {


            //构建put对象
            /**
             * 第一种方式
             * 以班级和学号一起作为rowkey
             * 查询的时候使用rowkey前缀过滤器
             *
             */
           /* Put put = new Put(key.toString().getBytes());
            put.add("i".getBytes(), "n".getBytes(), "".getBytes());*/

            /**
             *以班级作为rowkey   学号作为列名
             * 查询的时候直接get
             *
             */
            String[] split = key.toString().split("-");
            String clazz = split[0];
            String id = split[1];


            Put put = new Put(clazz.getBytes());
            put.add("i".getBytes(),id.getBytes(),"".getBytes());

            //将数据写入hbase
            context.write(NullWritable.get(), put);


        }
    }


    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();

        //指定zk连接地址
        conf.set("hbase.zookeeper.quorum", "node2:2181,node3:2181.node4:2181");

        Job job = Job.getInstance(conf);

        job.setJobName("HbaseToIndex");
        job.setJarByClass(Demo11HbaseToIndex.class);


        /**
         * 指定map端
         *
         */

        //可以指定过滤器
        Scan scan = new Scan();
        scan.addFamily("info".getBytes());

        TableMapReduceUtil.initTableMapperJob("student", scan, HbaseToIndexMap.class, Text.class, NullWritable.class, job);

        TableMapReduceUtil.initTableReducerJob("student_index", HbaseToIndexReduce.class, job);


        job.waitForCompletion(true);


    }

}
